package com.pxene.dmp.common;

import java.io.IOException;
import java.io.InputStream;
import java.util.Map;
import java.util.concurrent.CountDownLatch;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.http.HttpResponse;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.concurrent.FutureCallback;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
import org.apache.http.impl.nio.client.HttpAsyncClients;

import com.google.gson.Gson;
import com.google.gson.JsonElement;
import com.google.gson.JsonObject;
import com.pxene.dmp.constant.BaseConstant;

public class KVSender {
	private final CountDownLatch latch;
	private final CloseableHttpAsyncClient asyneClient;
	private final HttpGet request;
	private final Gson gson = new Gson();
	private final Log log = LogFactory.getLog(KVSender.class);
	
	private final StringBuilder sb = new StringBuilder();
	
	private int retry = 3;
	
	public KVSender(String url, Map<String, String> params, CountDownLatch latch) {
		request = new HttpGet(url);
		request.setConfig(RequestConfig.custom().setSocketTimeout(2000).setConnectTimeout(2000).build());
		for (Map.Entry<String, String> entry : params.entrySet()) {
			request.setHeader(entry.getKey(), entry.getValue());
			sb.append(entry.getKey()).append("=").append(entry.getValue()).append(",");
		}
		this.latch = latch;
		this.asyneClient = HttpAsyncClients.createDefault();
	}
	
	public void execute() {
		asyneClient.start();
		asyneClient.execute(request, new FutureCallback<HttpResponse>() {
	        public void completed(final HttpResponse resp) {
	            latch.countDown();
	            int code = resp.getStatusLine().getStatusCode();
	            if (code != 200) {
	            	log.error(BaseConstant.LOG_PREFIX + "KV server error:" + getMessage(resp) + "  " + sb.substring(0, sb.length()-1));
	            	reExecute(retry);
	            }
	        }
	        public void failed(final Exception ex) {
	        	latch.countDown();
	        	log.error(BaseConstant.LOG_PREFIX + "sendData exception:" + ex.getMessage());
	        	reExecute(retry);
	        }
	        public void cancelled() {
	        	latch.countDown();
	        }
	    });
	}
	
	private void reExecute(int retry) {
		CloseableHttpClient client = HttpClients.createDefault();
		try {
			for (int i=0; i<retry; i++) {
				HttpResponse resp = client.execute(request);
				if (resp.getStatusLine().getStatusCode() == 200) {
					break;
				} else {
					log.error(BaseConstant.LOG_PREFIX + "KV server error:" + getMessage(resp) + "  " + sb.substring(0, sb.length()-1));
				}
			}
		} catch (Exception e) {
			log.error(BaseConstant.LOG_PREFIX + "sendData exception:" + e.getMessage());
		} finally {
			close(client);
		}
	}
	
	private String getMessage(HttpResponse resp) {
		byte[] cache = new byte[512];
		try {
			InputStream input = resp.getEntity().getContent();
			input.read(cache);
		} catch (Exception e) {
			log.error(BaseConstant.LOG_PREFIX + "getMessage exception:" + e.getMessage());
		}
		JsonElement message = gson.fromJson(new String(cache, 0, cache.length).trim(), JsonObject.class).get("message");
		
		return message == null ? "" : message.getAsString();
	}
	
	public void close() {
		if (asyneClient != null && asyneClient.isRunning()) {
			try {
				asyneClient.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
	}
	
	private void close(CloseableHttpClient client) {
		if (client != null) {
			try {
				client.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
	}
	
//	public static void main(String[] args) throws ClientProtocolException, IOException {
//		CloseableHttpClient client = HttpClients.createDefault();
//		HttpGet request = new HttpGet("http://111.235.158.136:8080/bdapi/restful/fog/pxene/setUserInfo/ctyun_bdcsc_pxene/170699b99cadb6a63c5c2ffc898f5e0a");
//		Map<String, String> params = new HashMap<String, String>();
//		params.put("index", "102");
//		params.put("date", "2015101816");
//		params.put("adid", "12345");
//		params.put("type", "1");
//		params.put("imsi", "qwert");
//		params.put("meid", "zxcvb");
//		params.put("useragent", "iphone#adfas");
//		params.put("source", "baidu");
//		params.put("count", "100");
//		params.put("province", "BJ");
//		params.put("datatype", "2");
//		for (Map.Entry<String, String> entry : params.entrySet()) {
//			request.setHeader(entry.getKey(), entry.getValue());
//		}
//		client.execute(request);
//		client.close();
//	}
	
}
